use flutter_rust_bridge::frb;
// StreamSink 从生成的 frb_generated 模块导入
use crate::frb_generated::StreamSink;
use crate::frb_generated::{SseEncode, SseDecode};
use flutter_rust_bridge::for_generated::{SseSerializer, SseDeserializer};
use flutter_rust_bridge::for_generated::byteorder::{NativeEndian, WriteBytesExt, ReadBytesExt};
use std::io::Write;
use std::io::Read;
use serde::{Deserialize, Serialize};
use std::sync::{Arc, Mutex};
use tokio::sync::mpsc;
use tokio_tungstenite::{connect_async, tungstenite::Message as WsMessage};
use http::{Request, HeaderValue};
use futures_util::{SinkExt, StreamExt};
use anyhow::{Result, Context};
use url::Url;
use std::fmt;
use base64::{Engine as _, engine::general_purpose::STANDARD};

/// WebSocket 连接状态
#[derive(Debug, Clone, Serialize, Deserialize)]
#[frb]
pub enum WebSocketStatus {
    Disconnected,
    Connecting,
    Connected,
    Error(String),
}

// 手动实现 SseEncode 和 SseDecode（因为代码生成工具可能还没有生成）
impl SseEncode for WebSocketStatus {
    fn sse_encode(self, serializer: &mut SseSerializer) {
        // 使用 JSON 序列化
        let json = serde_json::to_string(&self).unwrap();
        let len = json.len() as i32;
        serializer.cursor.write_i32::<NativeEndian>(len).unwrap();
        serializer.cursor.write_all(json.as_bytes()).unwrap();
    }
}

impl SseDecode for WebSocketStatus {
    fn sse_decode(deserializer: &mut SseDeserializer) -> Self {
        let len = deserializer.cursor.read_i32::<NativeEndian>().unwrap() as usize;
        let mut buf = vec![0u8; len];
        deserializer.cursor.read_exact(&mut buf).unwrap();
        serde_json::from_slice(&buf).unwrap()
    }
}

/// WebSocket 消息
#[derive(Debug, Clone, Serialize, Deserialize)]
#[frb]
pub struct WebSocketMessage {
    pub data: String,
    pub message_type: String, // "text" | "binary" | "ping" | "pong"
}

// 手动实现 SseEncode 和 SseDecode
impl SseEncode for WebSocketMessage {
    fn sse_encode(self, serializer: &mut SseSerializer) {
        let json = serde_json::to_string(&self).unwrap();
        let len = json.len() as i32;
        serializer.cursor.write_i32::<NativeEndian>(len).unwrap();
        serializer.cursor.write_all(json.as_bytes()).unwrap();
    }
}

impl SseDecode for WebSocketMessage {
    fn sse_decode(deserializer: &mut SseDeserializer) -> Self {
        let len = deserializer.cursor.read_i32::<NativeEndian>().unwrap() as usize;
        let mut buf = vec![0u8; len];
        deserializer.cursor.read_exact(&mut buf).unwrap();
        serde_json::from_slice(&buf).unwrap()
    }
}

/// WebSocket 连接管理器
#[derive(Clone)]
#[frb(opaque)]
pub struct WebSocketManager {
    sender: Arc<Mutex<Option<mpsc::UnboundedSender<WsMessage>>>>,
    status_sink: Arc<Mutex<Option<StreamSink<WebSocketStatus>>>>,
    message_sink: Arc<Mutex<Option<StreamSink<WebSocketMessage>>>>,
    ping_handle: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
    read_handle: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
}

impl WebSocketManager {
    #[frb(sync)]
    pub fn new() -> Self {
        Self {
            sender: Arc::new(Mutex::new(None)),
            status_sink: Arc::new(Mutex::new(None)),
            message_sink: Arc::new(Mutex::new(None)),
            ping_handle: Arc::new(Mutex::new(None)),
            read_handle: Arc::new(Mutex::new(None)),
        }
    }

    /// 设置状态流
    pub fn set_status_sink(&self, sink: StreamSink<WebSocketStatus>) {
        *self.status_sink.lock().unwrap() = Some(sink);
    }

    /// 设置消息流
    pub fn set_message_sink(&self, sink: StreamSink<WebSocketMessage>) {
        *self.message_sink.lock().unwrap() = Some(sink);
    }

    /// 连接 WebSocket
    /// status_sink: 状态流，用于发送连接状态变化
    /// message_sink: 消息流，用于发送接收到的消息
    #[frb(sync)]
    pub fn connect(
        &self,
        subscription_id: String,
        ws_url: Option<String>,
        token: Option<String>,
        status_sink: StreamSink<WebSocketStatus>,
        message_sink: StreamSink<WebSocketMessage>,
    ) -> Result<()> {
        // 如果已经连接，先断开
        if self.sender.lock().unwrap().is_some() {
            self.disconnect();
        }

        // 设置流
        *self.status_sink.lock().unwrap() = Some(status_sink);
        *self.message_sink.lock().unwrap() = Some(message_sink);

        let default_url = format!("ws://127.0.0.1:3001/ws/{}", subscription_id);
        let url_str = ws_url.unwrap_or(default_url);

        // 通知连接中
        self.notify_status(WebSocketStatus::Connecting);

        // 在 Tokio runtime 中执行异步连接
        // 创建新的 runtime（因为这是在同步上下文中调用）
        let rt = tokio::runtime::Runtime::new().context("Failed to create Tokio runtime")?;
        
        // 在 runtime 中执行异步连接
        rt.block_on(async {
            match self.connect_async(&url_str, &token, &subscription_id).await {
                Ok(_) => {
                    eprintln!("✅ WebSocket connected successfully (Rust)");
                }
                Err(e) => {
                    eprintln!("❌ WebSocket connection failed: {}", e);
                    self.notify_status(WebSocketStatus::Error(e.to_string()));
                }
            }
        });

        Ok(())
    }

    async fn connect_async(
        &self,
        url_str: &str,
        token: &Option<String>,
        subscription_id: &str,
    ) -> Result<()> {
        eprintln!("🔌 Connecting to WebSocket (Rust): {}", url_str);
        eprintln!("🔑 Using subscription_id: {}", subscription_id);
        
        let url = url_str.parse::<Url>()
            .context("Invalid WebSocket URL")?;

        // 构建 Host 头部
        let host = match (url.host_str(), url.port(), url.scheme()) {
            (Some(host), Some(port), _) => {
                let is_standard_port = match url.scheme() {
                    "ws" | "http" => port == 80,
                    "wss" | "https" => port == 443,
                    _ => false,
                };
                if is_standard_port {
                    host.to_string()
                } else {
                    format!("{}:{}", host, port)
                }
            },
            (Some(host), None, _) => host.to_string(),
            (None, _, _) => return Err(anyhow::anyhow!("URL missing host")),
        };

        // 生成 Sec-WebSocket-Key
        use rand::RngCore;
        let mut key_bytes = [0u8; 16];
        rand::thread_rng().fill_bytes(&mut key_bytes);
        let sec_websocket_key = STANDARD.encode(&key_bytes);

        // 构建请求
        let mut request_builder = Request::builder()
            .uri(url.as_str())
            .header("Host", &host)
            .header("Upgrade", "websocket")
            .header("Connection", "Upgrade")
            .header("Sec-WebSocket-Key", &sec_websocket_key)
            .header("Sec-WebSocket-Version", "13");

        // 添加 Authorization header
        if let Some(t) = token {
            let auth_header = format!("Bearer {}", t);
            request_builder = request_builder.header(
                "Authorization",
                HeaderValue::from_str(&auth_header)
                    .context("Invalid Authorization header")?,
            );
        }

        let request = request_builder
            .body(())
            .context("Failed to build WebSocket request")?;

        // 连接 WebSocket
        eprintln!("🔗 Attempting WebSocket connection...");
        let (ws_stream, _) = connect_async(request)
            .await
            .context("WebSocket connection failed")?;
        
        eprintln!("✅ WebSocket handshake successful");

        let (mut write, mut read) = ws_stream.split();

        // 创建消息通道
        let (tx, mut rx) = mpsc::unbounded_channel::<WsMessage>();

        // 更新状态
        *self.sender.lock().unwrap() = Some(tx.clone());

        // 通知连接成功
        self.notify_status(WebSocketStatus::Connected);

        // 启动 ping 任务
        let sender_clone = Arc::clone(&self.sender);
        let status_sink_clone = Arc::clone(&self.status_sink);
        let mut ping_interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
        ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

        let ping_handle = tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = ping_interval.tick() => {
                        if let Some(sender) = sender_clone.lock().unwrap().as_ref() {
                            eprintln!("🏓 Sending WebSocket Ping");
                        if sender.send(WsMessage::Ping(vec![])).is_err() {
                                // Ping 失败，连接可能已断开
                                eprintln!("❌ Ping failed, connection may be dead");
                                *sender_clone.lock().unwrap() = None;
                                if let Some(sink) = status_sink_clone.lock().unwrap().as_ref() {
                                    sink.add(WebSocketStatus::Error("Ping failed".to_string()));
                                }
                                break;
                            }
                        } else {
                            break;
                        }
                    }
                }
            }
        });

        *self.ping_handle.lock().unwrap() = Some(ping_handle);

        // 启动读取任务
        let message_sink_clone = Arc::clone(&self.message_sink);
        let sender_clone = Arc::clone(&self.sender);
        let status_sink_clone = Arc::clone(&self.status_sink);

        let read_handle = tokio::spawn(async move {
            while let Some(msg_result) = read.next().await {
                match msg_result {
                    Ok(WsMessage::Text(text)) => {
                        eprintln!("📨 Received WebSocket text message: {} bytes", text.len());
                        if let Some(sink) = message_sink_clone.lock().unwrap().as_ref() {
                            sink.add(WebSocketMessage {
                                data: text,
                                message_type: "text".to_string(),
                            });
                        }
                    }
                    Ok(WsMessage::Binary(data)) => {
                        if let Some(sink) = message_sink_clone.lock().unwrap().as_ref() {
                            if let Ok(text) = String::from_utf8(data) {
                                sink.add(WebSocketMessage {
                                    data: text,
                                    message_type: "binary".to_string(),
                                });
                            }
                        }
                    }
                    Ok(WsMessage::Ping(_)) => {
                        // 自动回复 pong（tokio-tungstenite 会自动处理）
                    }
                    Ok(WsMessage::Pong(_)) => {
                        // 收到 pong，连接正常
                        eprintln!("🏓 Received WebSocket Pong, connection is alive");
                    }
                    Ok(WsMessage::Close(_)) => {
                        *sender_clone.lock().unwrap() = None;
                        if let Some(sink) = status_sink_clone.lock().unwrap().as_ref() {
                            sink.add(WebSocketStatus::Disconnected);
                        }
                        break;
                    }
                    Err(e) => {
                        *sender_clone.lock().unwrap() = None;
                        if let Some(sink) = status_sink_clone.lock().unwrap().as_ref() {
                            sink.add(WebSocketStatus::Error(format!("WebSocket error: {}", e)));
                        }
                        break;
                    }
                    _ => {}
                }
            }
        });

        *self.read_handle.lock().unwrap() = Some(read_handle);

        // 启动写入任务
        let sender_clone = Arc::clone(&self.sender);
        let status_sink_clone: Arc<Mutex<Option<StreamSink<WebSocketStatus>>>> = Arc::clone(&self.status_sink);
        tokio::spawn(async move {
            while let Some(msg) = rx.recv().await {
                if write.send(msg).await.is_err() {
                    *sender_clone.lock().unwrap() = None;
                    if let Some(sink) = status_sink_clone.lock().unwrap().as_ref() {
                        sink.add(WebSocketStatus::Error("Send failed".to_string()));
                    }
                    break;
                }
            }
        });

        Ok(())
    }

    /// 断开连接
    #[frb(sync)]
    pub fn disconnect(&self) {
        // 停止 ping 任务
        if let Some(handle) = self.ping_handle.lock().unwrap().take() {
            handle.abort();
        }

        // 停止读取任务
        if let Some(handle) = self.read_handle.lock().unwrap().take() {
            handle.abort();
        }

        // 关闭发送通道
        *self.sender.lock().unwrap() = None;

        self.notify_status(WebSocketStatus::Disconnected);
    }

    /// 发送消息
    #[frb(sync)]
    pub fn send(&self, message: String) -> Result<()> {
        let sender = self.sender.lock().unwrap();
        if let Some(tx) = sender.as_ref() {
            tx.send(WsMessage::Text(message))
                .map_err(|e| anyhow::anyhow!("Failed to send message: {}", e))?;
            Ok(())
        } else {
            Err(anyhow::anyhow!("WebSocket not connected"))
        }
    }

    /// 获取连接状态
    #[frb(sync)]
    pub fn get_status(&self) -> WebSocketStatus {
        if self.sender.lock().unwrap().is_some() {
            WebSocketStatus::Connected
        } else {
            WebSocketStatus::Disconnected
        }
    }

    fn notify_status(&self, status: WebSocketStatus) {
        if let Some(sink) = self.status_sink.lock().unwrap().as_ref() {
            sink.add(status);
        }
    }
}

impl Default for WebSocketManager {
    fn default() -> Self {
        Self::new()
    }
}

